package cn.com.receive.handler;

import cn.com.cache.CacheFactory;
import cn.com.cache.ZrCache;
import cn.com.common.Constant;
import cn.com.common.Request;
import cn.com.common.model.CacheValueModel;
import cn.com.common.model.ProbeModel;
import cn.com.rpc.channel.Channel;
import cn.com.rpc.handler.Handler;
import cn.com.rule.RuleDictionary;
import cn.com.common.RuleModel;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;

import java.util.concurrent.atomic.AtomicLong;

/**
 * 接收server端探测出来的热key
 *
 * @author jiaming
 */
public class ReceiveHotKeyHandler implements Handler {

    protected static final Logger log = LogManager.getLogger();

    private AtomicLong atomicLong = new AtomicLong(0);

    @Override
    public void connect(Channel channel, Object msg) {

    }

    @Override
    public void disconnect(Channel channel, Object msg) {

    }

    @Override
    public void received(Channel channel, Object msg) {
        if (msg instanceof Request) {
            Request request = (Request) msg;
            Object data = request.getData();
            if (data instanceof ProbeModel) {
                ProbeModel probeModel = (ProbeModel) data;
                if (log.isDebugEnabled()) {
                    log.debug("receive server push hotkey,[appName:{},keyType:{},key:{},keySum:{}]",
                            probeModel.getAppName(), probeModel.getKeyType(), probeModel.getKey(), probeModel.getKeySum());
                }
                String key = probeModel.getKey();
                RuleModel rule = RuleDictionary.findRule(key);
                if (rule == null) {
                    log.warn(" this key: [{}] is not find rule", key);
                    return;
                }
                //获取缓存组件
                ZrCache cache = CacheFactory.getNonNullCache(key);
                if (cache.get(key) != null) {
                    log.warn("pushed repeat hotkey,key:{}", key);
                }
                CacheValueModel cacheValueModel = new CacheValueModel(Constant.CACHE_DEFAULT_VALUE, rule.getDuration());
                //设置缓存
                cache.set(key, cacheValueModel);
                atomicLong.incrementAndGet();
                if (atomicLong.get() == 1L) {
                    log.info("received key number:{}", atomicLong.get());
                }
                if (atomicLong.get() % 10000L == 0L) {
                    log.info("received key number:{}", atomicLong.get());
                }
            }
        }
    }

    @Override
    public void sent(Channel channel, Object result) {

    }

    @Override
    public void cause(Channel channel, Throwable cause) {

    }
}
